package TableAndSQL

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{FileSystem, OldCsv, Schema}

object FileOutputTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //创建一个基于blink的流式
    val BlinkStreamSetting = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val blinkStreamTableEnv = StreamTableEnvironment.create(env, BlinkStreamSetting)

    //读取外部文件
    val filePath = "src/main/resources/SensorReading"
    //读取数据
    blinkStreamTableEnv.connect(new FileSystem().path(filePath))
      .withFormat(new OldCsv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temp", DataTypes.DOUBLE())
      ).createTemporaryTable("inputTable")

    val inputTable = blinkStreamTableEnv.from("inputTable")

    //转换操作
    //简单转换
    val resultTable = inputTable
      .select('id, 'temp)
      .filter('id === "sensor_1")

    //聚合转换
    val aggTable = inputTable
      .groupBy('id)
      .select('id, 'id.count as 'count)

    //输出到文件
    //注册输出表
    val fileOutPath = "src/main/resources/SensorReading_out"
    blinkStreamTableEnv.connect(new FileSystem().path(fileOutPath))
      .withFormat(new OldCsv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("temperature", DataTypes.DOUBLE())
      ).createTemporaryTable("outputTable")

    //    resultTable.toAppendStream[(String, Double)].print()
    //    aggTable.toRetractStream[(String, Long)].print("agg")

    aggTable.insertInto("outputTable")

    env.execute()
  }
}
